// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::any::Any;

use api::v1::value::ValueData;
use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType};
use common_event_recorder::error::Result;
use common_event_recorder::Event;
use common_time::timestamp::{TimeUnit, Timestamp};

use crate::{ProcedureId, ProcedureState};

pub const EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME: &str = "procedure_id";
pub const EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME: &str = "procedure_state";
pub const EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME: &str = "procedure_error";

/// `ProcedureEvent` represents an event emitted by a procedure during its execution lifecycle.
#[derive(Debug)]
pub struct ProcedureEvent {
    /// Unique identifier associated with the originating procedure instance.
    pub procedure_id: ProcedureId,
    /// The timestamp of the event.
    pub timestamp: Timestamp,
    /// The state of the procedure.
    pub state: ProcedureState,
    /// The event emitted by the procedure. It's generated by [Procedure::event].
    pub internal_event: Box<dyn Event>,
}

impl ProcedureEvent {
    pub fn new(
        procedure_id: ProcedureId,
        internal_event: Box<dyn Event>,
        state: ProcedureState,
    ) -> Self {
        Self {
            procedure_id,
            internal_event,
            timestamp: Timestamp::current_time(TimeUnit::Nanosecond),
            state,
        }
    }
}

impl Event for ProcedureEvent {
    fn event_type(&self) -> &str {
        self.internal_event.event_type()
    }

    fn timestamp(&self) -> Timestamp {
        self.timestamp
    }

    fn json_payload(&self) -> Result<String> {
        self.internal_event.json_payload()
    }

    fn extra_schema(&self) -> Vec<ColumnSchema> {
        let mut schema = vec![
            ColumnSchema {
                column_name: EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME.to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            },
            ColumnSchema {
                column_name: EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME.to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            },
            ColumnSchema {
                column_name: EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME.to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            },
        ];
        schema.append(&mut self.internal_event.extra_schema());
        schema
    }

    fn extra_row(&self) -> Result<Row> {
        let error_str = match &self.state {
            ProcedureState::Failed { error } => format!("{:?}", error),
            ProcedureState::PrepareRollback { error } => format!("{:?}", error),
            ProcedureState::RollingBack { error } => format!("{:?}", error),
            ProcedureState::Retrying { error } => format!("{:?}", error),
            ProcedureState::Poisoned { error, .. } => format!("{:?}", error),
            _ => "".to_string(),
        };
        let mut row = vec![
            ValueData::StringValue(self.procedure_id.to_string()).into(),
            ValueData::StringValue(self.state.as_str_name().to_string()).into(),
            ValueData::StringValue(error_str).into(),
        ];
        row.append(&mut self.internal_event.extra_row()?.values);
        Ok(Row { values: row })
    }

    fn as_any(&self) -> &dyn Any {
        self
    }
}
